package storm2;// -*- codeing: utf-8 -*-

// @Time     :2021/12/10 10:43
// @Author   :郭荣杰
//@File      :Bolt1_orderdetail.java
//@Software  : IntelliJ IDEA
import net.sf.json.JSONArray;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import net.sf.json.JSONObject;

public class Bolt1_orderdetail extends BaseBasicBolt{
    public void execute(Tuple input, BasicOutputCollector collector) {
        // TODO Auto-generated method stub
        /**
         * kafkaSpout发送的字段名为bytes
         */

        /**
         * 描述字段
         */
        /**-----------------------------------------------
         * 1.orderdetailid	    varchar(50)     订单详情 ID
         * 2.orderid	        varchar(50)     订单唯一 ID
         * 3.sender	            varchar(10)     发货人姓名
         * 4.sendplace	        varchar(50)     发货地址
         * 5.recipient	        varchar(10)     收货人姓名
         * 6.receiveplace	    varchar(50)     收货地址
         * 7.methodofpayment	varchar(50)     支付方式
         * 8.paytime	        varchar(50)     支付时间
         * 9.freight	        float           运费
         * 10.total	            float           订单总金额
         */

        try {
            //得到字节码
            byte[]bytes=input.getBinary(0);  //字节码
            //byte->String
            //192.168.53.123	0-001	100	手机	小米8	99	2	20181112（timeStamp）
            //字节码转字符串
            String line =new String(bytes);
            //声明一个JSONObejct对象
            JSONObject jsonObject=new net.sf.json.JSONObject();
            //声明一个JSONArray对象
            JSONArray jsonArray =new JSONArray();
            //利用jsonObject方法将字符串转JSON
            JSONObject json_line = jsonObject.fromObject(line);
            //得到数据data，去除元数据
            JSONArray data =jsonArray.fromObject(json_line.getString("data"));
            //得到数据data各字段的值
//        System.out.println(data.toArray()[0]);
//            String orderdetailid =jsonObject.fromObject(data.toArray()[0]).getString("orderdetailid");
//            String orderid=jsonObject.fromObject(data.toArray()[0]).getString("orderid");
//            String sender =jsonObject.fromObject(data.toArray()[0]).getString("sender");
            String sendplace =jsonObject.fromObject(data.toArray()[0]).getString("sendplace");
            String recipient=jsonObject.fromObject(data.toArray()[0]).getString("recipient");
            String receiveplace=jsonObject.fromObject(data.toArray()[0]).getString("receiveplace");
            String methodofpayment=jsonObject.fromObject(data.toArray()[0]).getString("methodofpayment");
//            String paytime=jsonObject.fromObject(data.toArray()[0]).getString("paytime");
//            double freight=jsonObject.fromObject(data.toArray()[0]).getDouble("freight");
            double total=jsonObject.fromObject(data.toArray()[0]).getDouble("total");
            collector.emit(new Values(sendplace,recipient,receiveplace,
                    methodofpayment,total));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        /**
         * 描述字段
         */
        /**-----------------------------------------------
         * 1.orderdetailid	    varchar(50)     订单详情 ID
         * 2.orderid	        varchar(50)     订单唯一 ID
         * 3.sender	            varchar(10)     发货人姓名
         * 4.sendplace	        varchar(50)     发货地址
         * 5.recipient	        varchar(10)     收货人姓名
         * 6.receiveplace	    varchar(50)     收货地址
         * 7.methodofpayment	varchar(50)     支付方式
         * 8.paytime	        varchar(50)     支付时间
         * 9.freight	        float           运费
         * 10.total	            float           订单总金额
         */

        declarer.declare(new Fields("sendplace","recipient","receiveplace","methodofpayment","total"));
    }

}